package com.atguigu.day04;

import com.atguigu.bean.Event;
import com.mysql.jdbc.Driver;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

public class Flink02_Sink_JDBC_Mysql {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.获取数据
//        DataStreamSource<Event> streamSource = env.fromElements(
//                new Event("Songsong", "./home", 1000L),
//                new Event("Bingbing", "./cart", 2000L)
//                );

        SingleOutputStreamOperator<Event> streamSource = env.socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Event>() {
                    @Override
                    public Event map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new Event(split[0], split[1], Long.parseLong(split[2]));
                    }
                });

        //TODO 3.将数据写入Mysql
        SinkFunction<Event> sink = JdbcSink.sink(
                "insert into clicks (user, url) VALUES (?, ?)",
                new JdbcStatementBuilder<Event>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                        preparedStatement.setString(1, event.user);
                        preparedStatement.setString(2, event.url);
                    }
                },
                JdbcExecutionOptions.builder()
                        //设置批次大小为1 代表的含义是 来一条数据就进行刷写
                        .withBatchSize(1)
                        .build()
                , new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://hadoop102:3306/test")
                        .withUsername("root")
                        .withPassword("123456")
                        .withDriverName(Driver.class.getName())
                        .build());

        streamSource.addSink(sink);

        env.execute();
    }
}
